package com.xian.scala.stream

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.shaded.netty4.io.netty.util.internal.StringUtil
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala._

object SocketWC {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val lines: DataStream[String] = env.socketTextStream("localhost", 2323)

    //lines.print()

    //过滤空字符串
    val works: DataStream[String] = lines.filter(line => line.nonEmpty).flatMap(_.split(" "))


    val work_one: DataStream[(String, Int)] = works.map(work => (work, 1))

    val group_work_one: KeyedStream[(String, Int), Tuple] = work_one.keyBy(0)

    val work_sum: DataStream[(String, Int)] = group_work_one.sum(1)

    work_sum.print()


    //执行
    env.execute()

  }

}
